// (c) 2007 Messaging Logic. All Rights Reserved.
package org.xi8ix.jms;

import javax.jms.JMSException;
import javax.jms.Message;
import javax.jms.MessageConsumer;
import javax.jms.MessageListener;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;
import java.util.logging.Level;
import java.util.logging.Logger;

/**
 * @author Iain Shigeoka
 */
public class JMSMessageConsumer implements MessageConsumer {
    private static Logger LOG = Logger.getLogger(JMSMessageConsumer.class.getName());
    private JMSDestination destination;
    private BlockingQueue<Message> queue = new LinkedBlockingQueue<Message>();

    public JMSMessageConsumer(JMSDestination destination) {
        this.destination = destination;
        destination.add(queue);
    }

    public String getMessageSelector() throws JMSException {
        return null;
    }

    public MessageListener getMessageListener() throws JMSException {
        return null;
    }

    public void setMessageListener(MessageListener messageListener) throws JMSException {
    }

    public Message receive() throws JMSException {
        try {
            LOG.log(Level.FINEST,"Receive wait {0}",destination);
            Message msg = queue.take();
            LOG.log(Level.FINEST,"Received {0}",msg);
            return msg;
        } catch (InterruptedException e) {
            throw new JMSException(e.getMessage());
        }
    }

    public Message receive(long l) throws JMSException {
        try {
            LOG.log(Level.FINEST,"Receive waiting {0} on {1}",new Object[]{l,destination});
            Message msg = queue.poll(l, TimeUnit.MILLISECONDS);
            LOG.log(Level.FINEST,"Received {0}",msg);
            return msg;
        } catch (InterruptedException e) {
            throw new JMSException(e.getMessage());
        }
    }

    public Message receiveNoWait() throws JMSException {
        LOG.log(Level.FINEST,"Recv no wait {0}",destination);
        Message msg = queue.poll();
        LOG.log(Level.FINEST,"Received {0}",msg);
        return msg;
    }

    public void close() throws JMSException {
        LOG.log(Level.FINEST,"Closing consumer {0}",destination);
        destination.remove(queue);
        LOG.log(Level.FINEST,"Consumer closed");
    }
}
